#include "producer_mqtt.h"

#include <queue>
#ifdef WIN32
#define usleep(x) Sleep(x)
#endif
#include "business_def.h"
#include "Log.h"
#include "py_mqtt.h"

ProducerMQTT::ProducerMQTT(PyMQTTIO *mqttcpp_)
	: running(true)
	, mqttcpp(mqttcpp_)
{
	pub_s = "P" + mqttfunc.getAreaID();	//重设发布主题,后续会支持从配置文件读取
}

ProducerMQTT::~ProducerMQTT()
{
	running = false;
}

void* ProducerMQTT::run()
{
	while (running)
	{
		if (NULL!= mqttcpp&&mqttcpp->link) {
			send_pval();
			send_reply();
			usleep(100);
		}
	}
	return 0;
}

void ProducerMQTT::send_pval()
{
	std::string pvJsonDesc = "";
	mqttfunc.getPValueJsonDesc(pvJsonDesc);
	/*
	*demo
	*pvJsonDesc =
	* "{ \"area_sn\":\"1075605\", \"dev_sn\" : \"3\",\"point_sn\" : \"2\",\"value\" : \"1\",\"code\" : \"test\" }";
	*/
	if (!pvJsonDesc.empty())
	{
		sendMsg(pvJsonDesc.c_str(), static_cast<int>(pvJsonDesc.length()));
	}
}

void ProducerMQTT::send_reply()
{
	std::string m_ReplyText_s = "";
	if (mqttcpp->pop_rep(m_ReplyText_s))
	{
		sendMsg(m_ReplyText_s.c_str(), static_cast<int>(m_ReplyText_s.length()));
#ifdef _DEBUG
		printf("return send[%d]:\n", (int)m_ReplyText_s.length());
		for (int i = 0; i < m_ReplyText_s.length(); i++)
		{
			printf("%02hhx ", (char)(m_ReplyText_s.at(i)));
		}
		printf("\n");
#endif // DEBUG
	}
}

bool ProducerMQTT::sendMsg(const char* buf, int len)
{
	bool re = true;
	try {
		int ret = 0;
		//ret = mqttcpp->loop_write();
		//if (MOSQ_ERR_SUCCESS != ret)
		//{
		//	mqttcpp->link = false;
		//}
		ret = mqttcpp->publish(NULL, pub_s.c_str(), len, buf);
		if (MOSQ_ERR_SUCCESS != ret)
		{
			mqttcpp->link = false;
			re = false;
			Print_WARN("mqttcpp publish error(%d)\n",ret);
		}
		ret = mqttcpp->loop_write();
		if (MOSQ_ERR_SUCCESS != ret)
		{
			mqttcpp->link = false;
			re = false;
			Print_WARN("mqttcpp loop_write error(%d)\n", ret);
		}
	}
	catch (...) {
		mqttcpp->link = false;
		re = false;
		Print_WARN("ProducerMQTT sendMsg error\n");
	}
	return re;
}